package com.lianxi.rocketmqmvc;


import com.lianxi.rocketmqmvc.transcation.TransactionListenerImpl;
import com.lianxi.util.ResourcesUtil;
import org.apache.commons.io.Charsets;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Description : rocket 生产者工具类
 *
 * @author : fy
 *         Date : 2018/1/24 16:45
 */
public class Producer {
    private static Logger logger = LoggerFactory.getLogger(Producer.class);
    /**
     * 默认超时时间
     */
    private static final Integer DEFAULT_TIME_OUT = Integer.parseInt(ResourcesUtil.getRocketTimeout());

    /**
     * 默认最大不压缩信息体积
     */
    private static final Integer DEFAULT_MSG_BODY_SIZE = Integer.parseInt(ResourcesUtil.getRocketMsgBodyOverHowMuch());

    /**
     * 队列数
     */
    private static final Integer DEFAULT_TOPIC_QUEUE_NUMS = Integer.parseInt(ResourcesUtil.getRocketMQTopicQueueNums());

    /**
     * 消息最大大小
     */
    private static final Integer DEFAULT_MESSAGE_MAX_SIZE = Integer.parseInt(ResourcesUtil.getRocketMQMessageMaxSize());

    /**
     * nameServer 地址
     */
    private static final String SRV_ADDR = ResourcesUtil.getRocketSrvAddr();

    private static DefaultMQProducer logProducer;

    private static DefaultMQProducer staticProducer;

    private static DefaultMQProducer staticBatchProducer;

    private static DefaultMQProducer siteProducer;

    private static DefaultMQProducer channelDelProducer;

    private static DefaultMQProducer coutProducer;

    private Producer() {

    }


    /**
     * api统计接口单例
     */
    private static synchronized DefaultMQProducer getCoutProducerInstance() {
        if (coutProducer == null) {
            synchronized (Producer.class) {
                if (coutProducer == null) {
                    coutProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_COUNT_GROUP);
                }
            }
        }
        return coutProducer;
    }

    /**
     * 站点生产者单例
     */
    private static synchronized DefaultMQProducer getChannelDelProducerInstance() {
        if (siteProducer == null) {
            synchronized (Producer.class) {
                if (siteProducer == null) {
                    siteProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_CHANNEL_DEL_PRODUCER_GROUP);
                }
            }
        }
        return siteProducer;
    }




    /**
     * 站点生产者
     */
    public static void startChannelDelProducer() {
        siteProducer = Producer.getChannelDelProducerInstance();
        siteProducer.setNamesrvAddr(SRV_ADDR);
        siteProducer.setVipChannelEnabled(false);
        siteProducer.setDefaultTopicQueueNums(DEFAULT_TOPIC_QUEUE_NUMS);
        siteProducer.setSendMsgTimeout(DEFAULT_TIME_OUT);
        siteProducer.setCompressMsgBodyOverHowmuch(DEFAULT_MSG_BODY_SIZE);
        try {
            //调用start()方法启动一个producer实例
            siteProducer.start();
            System.out.println(" --------------------- channelDelProducer start ---------------------");
            logger.info(" --------------------- channelDelProducer start ---------------------");
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }

    /**
     * 日志生产者单例
     */
    private static synchronized DefaultMQProducer getLogProducerInstance() {
        if (logProducer == null) {
            synchronized (Producer.class) {
                if (logProducer == null) {
                    logProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_LOG_PRODUCER_GROUP);
                }
            }
        }
        return logProducer;
    }

    /**
     * 启动日志生产者
     */
    public static void startLogProducer() {
        logProducer = Producer.getLogProducerInstance();
        logProducer.setNamesrvAddr(SRV_ADDR);
        logProducer.setVipChannelEnabled(false);
        logProducer.setDefaultTopicQueueNums(DEFAULT_TOPIC_QUEUE_NUMS);
        logProducer.setSendMsgTimeout(DEFAULT_TIME_OUT);
        logProducer.setCompressMsgBodyOverHowmuch(DEFAULT_MSG_BODY_SIZE);
        try {
            //调用start()方法启动一个producer实例
            logProducer.start();
            logger.info(" --------------------- logProducer start ---------------------");
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }



    /**
     * 站点生产者单例
     */
    private static synchronized DefaultMQProducer getSiteProducerInstance() {
        if (siteProducer == null) {
            synchronized (Producer.class) {
                if (siteProducer == null) {
                    siteProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_SITE_PRODUCER_GROUP);
                }
            }
        }
        return siteProducer;
    }

    /**
     * 站点生产者
     */
    public static void startSiteProducer() {
        siteProducer = Producer.getSiteProducerInstance();
        siteProducer.setNamesrvAddr(SRV_ADDR);
        siteProducer.setVipChannelEnabled(false);
        siteProducer.setDefaultTopicQueueNums(DEFAULT_TOPIC_QUEUE_NUMS);
        siteProducer.setSendMsgTimeout(DEFAULT_TIME_OUT);
        siteProducer.setCompressMsgBodyOverHowmuch(DEFAULT_MSG_BODY_SIZE);
        try {
            //调用start()方法启动一个producer实例
            siteProducer.start();
            logger.info(" --------------------- siteProducer start ---------------------");
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }

    /**
     * Description : 日志生产者推送message到broker
     *
     * @param msg 消息
     * @author : fy
     * Date : 2018/1/24 16:34
     */
    public static void sendLogMsg(String msg) throws Exception {
        //logger.info(" ------ topic : {} |  tag : {} 日志消息发送 ------ ", RocketMQConstant.ROCKETMQ_LOG_TOPIC, RocketMQConstant.ROCKETMQ_LOG_TAG);
        Message message = new Message(RocketMQConstant.ROCKETMQ_LOG_TOPIC,
                RocketMQConstant.ROCKETMQ_CMS_TAG, msg.getBytes(Charsets.UTF_8));
        logProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                sendSuccess(sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                logger.info(" rocket producer exception : {}", throwable.getMessage());
            }
        });
    }

    //======================================================== 全站生成 begin =========================================================

    /**
     * 静态页生产者单例（全站生成）
     */
    private static synchronized DefaultMQProducer getStaticBatchProducerInstance() {
        if (staticBatchProducer == null) {
            synchronized (Producer.class) {
                if (staticBatchProducer == null) {
                    staticBatchProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_STATIC_BATCH_PRODUCER_GROUP);
                }
            }
        }
        return staticBatchProducer;
    }

    /**
     * 启动静态页面生产者（全站生成）
     */
    public static void startStaticBatchProducer() {
        staticBatchProducer = Producer.getStaticBatchProducerInstance();
        //name service addr
        staticBatchProducer.setNamesrvAddr(SRV_ADDR);
        //关闭vip通道
        staticBatchProducer.setVipChannelEnabled(false);
        //一条消息的发送超时时间
        staticBatchProducer.setSendMsgTimeout(DEFAULT_TIME_OUT);
        //消息没有存储成功是否发送到另外一个broker.
        staticBatchProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
        //默认4M
        staticBatchProducer.setMaxMessageSize(DEFAULT_MESSAGE_MAX_SIZE);
        staticBatchProducer.setCompressMsgBodyOverHowmuch(DEFAULT_MSG_BODY_SIZE);
        try {
            staticBatchProducer.start();
            logger.info(" --------------------- staticBatchProducer start ---------------------");
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }

    private static synchronized DefaultMQProducer getResourceProducerInstance() {
        if (staticProducer == null) {
            synchronized (Producer.class) {
                if (staticProducer == null) {
                    staticProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_GROUP_IMPORTDATASOURCE);
                }
            }
        }
        return staticProducer;
    }


    //==================================================== 普通消费队列 begin ========================================================

    /**
     * 静态页生产者单例
     */
    private static synchronized DefaultMQProducer getStaticProducerInstance() {
        if (staticProducer == null) {
            synchronized (Producer.class) {
                if (staticProducer == null) {
                    staticProducer = new DefaultMQProducer(RocketMQConstant.ROCKETMQ_STATIC_PRODUCER_GROUP);
                }
            }
        }
        return staticProducer;
    }

    /**
     * 启动静态页面生产者
     */
    public static void startStaticProducer() {
        staticProducer = Producer.getStaticProducerInstance();
        //name service addr
        staticProducer.setNamesrvAddr(SRV_ADDR);
        //关闭vip通道
        staticProducer.setVipChannelEnabled(false);
        //一条消息的发送超时时间
        staticProducer.setSendMsgTimeout(DEFAULT_TIME_OUT);
        //消息没有存储成功是否发送到另外一个broker.
        staticProducer.setRetryAnotherBrokerWhenNotStoreOK(true);
        staticProducer.setRetryTimesWhenSendFailed(3);
        //默认4M
        staticProducer.setMaxMessageSize(DEFAULT_MESSAGE_MAX_SIZE);
        staticProducer.setCompressMsgBodyOverHowmuch(DEFAULT_MSG_BODY_SIZE);
        try {
            staticProducer.start();
            logger.info(" --------------------- staticProducer start ---------------------");
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }

    //==================================================== common ========================================================

    /**
     * 关闭生产者
     *
     * @param type 生产者队列类型 0 普通 1 全站生成 2 日志
     */
    public static void closeProducer(int type) {
        switch (type) {
            case RocketMQConstant.ROCKET_TYPE_NORMAL:
                staticProducer.shutdown();
                break;
            case RocketMQConstant.ROCKET_TYPE_BATCH:
                staticBatchProducer.shutdown();
                break;
            case RocketMQConstant.ROCKET_TYPE_LOG:
                logProducer.shutdown();
                break;
            case RocketMQConstant.ROCKET_TYPE_SITE:
                siteProducer.shutdown();
                break;
            case RocketMQConstant.ROCKET_TYPE_CHANNEL_DEL:
                channelDelProducer.shutdown();
                break;
            default:
        }
    }


//    /**
//     * Description : 静态页面生产者推送message到broker
//     *
//     * @param topic 消息队列分类
//     * @param tags  消息标签
//     * @param keys  消息标识  站点id + _ +新闻id
//     * @param msg   消息
//     * @param type  0 普通队列， 1 全站生成队列
//     * @author : fy
//     * Date : 2018/1/24 16:34
//     */
//    public static void sendStaticMsg(int type, String topic, String tags, String keys, String msg) throws Exception {
//        logger.info(" ------ topic : {} | tags : {} | keys : {} ------ ", topic, tags, keys);
//        Message message = new Message(topic, tags, keys, msg.getBytes(Charsets.UTF_8));
//        if (type == RocketMQConstant.ROCKET_TYPE_NORMAL) {
//            staticProducer.send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    sendSuccess(sendResult);
//                }
//                @Override
//                public void onException(Throwable throwable) {
//                    logger.error(" rocket producer exception : {}", throwable.getMessage());
//                }
//            });
//        } else {
//            staticBatchProducer.send(message, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    sendSuccess(sendResult);
//                }
//                @Override
//                public void onException(Throwable throwable) {
//                    logger.info(" rocket producer exception : {}", throwable.getMessage());
//                }
//            });
//        }
//    }




    public static void sendSiteMsg(String msg) throws Exception {
        Message message = new Message(RocketMQConstant.ROCKETMQ_SITE_TOPIC,  msg.getBytes(Charsets.UTF_8));
            siteProducer.send(message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    sendSuccess(sendResult);
                }
                @Override
                public void onException(Throwable throwable) {
                    logger.error(" rocket siteProducer exception : {}", throwable.getMessage());
                }
            });
    }

    public static void sendChannelDelMsg(String msg) throws Exception {
        Message message = new Message(RocketMQConstant.ROCKETMQ_CHANNEL_DEL_TOPIC,  msg.getBytes(Charsets.UTF_8));
        channelDelProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                sendSuccess(sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                logger.error(" rocket channelDelProducer exception : {}", throwable.getMessage());
            }
        });
    }


    private static void sendSuccess(SendResult sendResult) {
        SendStatus sendStatus = sendResult.getSendStatus();
        if(!"SEND_OK".equals(sendStatus)){
            MessageQueue messageQueue = sendResult.getMessageQueue();
            logger.info(" ######## send status : {}, | broker name : {}, | topic : {}, | queue id : {} ######## ",
                    sendStatus, messageQueue.getBrokerName(), messageQueue.getTopic(), messageQueue.getQueueId());
        }
    }



    /**
     * api统计接口
     */
    public static void startCoutProducerProducer() {
        coutProducer = Producer.getChannelDelProducerInstance();
        coutProducer.setNamesrvAddr(SRV_ADDR);
        coutProducer.setVipChannelEnabled(false);
        coutProducer.setDefaultTopicQueueNums(DEFAULT_TOPIC_QUEUE_NUMS);
        coutProducer.setSendMsgTimeout(DEFAULT_TIME_OUT);
        coutProducer.setCompressMsgBodyOverHowmuch(DEFAULT_MSG_BODY_SIZE);
        try {
            //调用start()方法启动一个producer实例
            coutProducer.start();
            System.out.println(" --------------------- count Producer start ---------------------");
        } catch (MQClientException e) {
            logger.info(e.getMessage());
        }
    }

    public static void sendCountMsg(String msg) throws Exception {
        Message message = new Message(RocketMQConstant.ROCKETMQ_COUNT_TOPIC, "tag-api", msg.getBytes(Charsets.UTF_8));
        //延时消息设置，一共如下18个等级，level=3代表十秒之后消费。
        //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        message.setDelayTimeLevel(3);
        coutProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                sendSuccess(sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                logger.error(" rocket siteProducer exception : {}", throwable.getMessage());
            }
        });
        //
    }


    /**
     * 开启事务性生产者
     * @param msg
     * @throws Exception
     */
    public static void sendTranscationCountMsg(String msg) throws Exception {
        //创建事务监听器
        TransactionListener transactionListener = new TransactionListenerImpl();
        Message message = new Message(RocketMQConstant.ROCKETMQ_COUNT_TOPIC, "tag-api", msg.getBytes(Charsets.UTF_8));
        //延时消息设置，一共如下18个等级，level=3代表十秒之后消费。
        //private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        message.setDelayTimeLevel(3);
        coutProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                sendSuccess(sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                logger.error(" rocket siteProducer exception : {}", throwable.getMessage());
            }
        });
        //
    }



}
